package com.telit.communicate.communicate;

import cn.hutool.json.JSONUtil;
import com.telit.common.util.StringUtils;
import com.telit.communicate.constant.CommunicateVariable;
import com.telit.communicate.constant.MsgSource;
import com.telit.communicate.constant.SystemName;
import com.telit.communicate.dto.MsgInfoDTO;
import com.telit.communicate.dto.UpAndDownLineDTO;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.lang.reflect.Field;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;


/**
 * @author zhouyu
 * @description handler业务处理类
 **/
@Component
@Slf4j
public class ServerHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    static {
        UserInstanceCache.createInstanceCache("channelUserInfo");
        UserInstanceCache.createInstanceCache("userIsFirstConnect");
    }


    /**
     * channel关联AttributeKey
     */
    public final static AttributeKey<String> KEY = AttributeKey.valueOf("userId");

    /**
     * 根据channel查询userid
     *
     * @param channel 唯一channel
     * @return userid
     */
    public static String getUserIdByChannel(Channel channel) {
        return channel.attr(KEY).get();
    }

    /**
     * 关联 userId 与 channel
     *
     * @param channel 通道
     * @param userId  userid
     */
    public static void setUserIdToChannel(Channel channel, String userId) {
        channel.attr(KEY).set(userId);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame) throws Exception {

    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        String userId = getUserIdByChannel(channel);
        Object value = UserInstanceCache.getValueFromCache("userIsFirstConnect", channel);
        if (value != null) {
            boolean isFirstConnect = (boolean) value;
            if (!isFirstConnect) {
                super.channelReadComplete(ctx);
                return;
            }
        }
        UpAndDownLineDTO upAndDownLineDTO = UpAndDownLineDTO.builder().userId(userId.split("_")[0]).upAndDown(0).build();
        sendMsgToCache(userId, upAndDownLineDTO);
        UserInstanceCache.setValueToCache("userIsFirstConnect", channel, false);
        super.channelReadComplete(ctx);
    }


    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        Channel channel = ctx.channel();
        //first connect is FullHttpRequest
        String userId;
        if (msg instanceof FullHttpRequest) {
            FullHttpRequest request = (FullHttpRequest) msg;
            String[] uri = request.uri().split("/");
            if (uri.length > 2) {
                userId = uri[2];
                setUserIdToChannel(channel, userId);
                UserInstanceCache.setValueToCache("channelUserInfo", channel, userId);
            }
        } else if (msg instanceof TextWebSocketFrame) {
            userId = getUserIdByChannel(channel);
            String[] path = userId.split("_");
            String receiveMsgKey = path[2] + path[1];
            log.info("receiveMsgKey：{}", receiveMsgKey);
            log.info("enableReceiveMsg：{}", String.join(",", CommunicateVariable.enableReceiveMsg.keySet()));
            // 2022-04-28 fix by tracine: 包装类布尔类型为空会引发 NullPointerException
            //if (CommunicateVariable.enableReceiveMsg.get(path[2] + path[1])) {
            if (Boolean.TRUE.equals(CommunicateVariable.enableReceiveMsg.get(receiveMsgKey))) {
                TextWebSocketFrame frame = (TextWebSocketFrame) msg;
                String receiveMsg = frame.text();
                MsgInfoDTO msgInfoDTO = MsgInfoDTO.builder().userId(path[0]).msg(receiveMsg).build();
                sendMsgToCache(userId, msgInfoDTO);
            } else {
                log.info("Message processing has stopped and cannot continue receiving data.....");
                log.info("Data not received is : " + ((TextWebSocketFrame) msg).text());
            }
        }
        super.channelRead(ctx, msg);
    }

    /**
     * 待发送消息放入缓存
     *
     * @param userId 用户id
     * @param dto    待发送消息
     */
    private void sendMsgToCache(String userId, Object dto) throws InterruptedException {
        String[] path = userId.split("_");
        if (StringUtils.checkUserId(userId, "ALL")) {
            MsgInstanceCache.setMsgInfo(MsgSource.WEB, SystemName.getSystemName(path[2]), dto);
            MsgInstanceCache.setMsgInfo(MsgSource.UAV, SystemName.getSystemName(path[2]), dto);
            MsgInstanceCache.setMsgInfo(MsgSource.PHONE, SystemName.getSystemName(path[2]), dto);
        } else {
            for (Field field : MsgSource.getFields()) {
                if (StringUtils.checkUserId(userId, field.getName())) {
                    MsgInstanceCache.setMsgInfo(MsgSource.getMsgSource(field.getName()), SystemName.getSystemName(path[2]), dto);
                }
            }
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        String userId = getUserIdByChannel(channel);
        UpAndDownLineDTO upAndDownLineDTO = UpAndDownLineDTO.builder().userId(userId.split("_")[0]).upAndDown(1).build();
        sendMsgToCache(userId, upAndDownLineDTO);
        UserInstanceCache.removeCacheByKey("channelUserInfo", ctx.channel());
        UserInstanceCache.removeCacheByKey("userIsFirstConnect", channel);
        super.channelInactive(ctx);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        UserInstanceCache.removeCacheByKey("channelUserInfo", ctx.channel());
        UserInstanceCache.removeCacheByKey("userIsFirstConnect", ctx.channel());
        super.exceptionCaught(ctx, cause);
    }

/*    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        //连接闲置处理
        Channel channel = ctx.channel();
        String userId = getUserIdByChannel(channel);
        if (userId != null && evt instanceof IdleStateEvent && !userId.contains(MsgSource.WEB.name())) {
            IdleState state = ((IdleStateEvent) evt).state();
            if (state == IdleState.ALL_IDLE) {
                ChannelFuture future = ctx.close();
                future.addListener((ChannelFutureListener) f -> {
                    UserInstanceCache.removeCacheByKey("channelUserInfo", channel);
                    UserInstanceCache.removeCacheByKey("userIsFirstConnect", channel);
                });
            }
        }
    }*/

    /**
     * 推送消息到所有客户端
     *
     * @param message 消息内容
     */
    public static void sendMessageToAll(String message) {
        List<ConcurrentHashMap<Object, Object>> channelUserInfoInstance = UserInstanceCache.getCacheInstance("channelUserInfo");
        channelUserInfoInstance.forEach(
                l -> {
                    l.forEach((key, value) -> {
                        Channel channel = (Channel) key;
                        sendMessage(channel, message);
                    });
                }
        );
    }

    /**
     * 推送消息到所有web端
     *
     * @param message 消息内容
     */
    public static void sendMessageToWeb(String message) {
        log.info("正在往web短推送消息 ---start");
        List<ConcurrentHashMap<Object, Object>> channelUserInfoInstance = UserInstanceCache.getCacheInstance("channelUserInfo");
        log.info("正在往web短推送消息 --- 展示实例对象[{}]", JSONUtil.toJsonStr(channelUserInfoInstance));
        channelUserInfoInstance.forEach(
                l -> {
                    l.forEach((key, value) -> {
                        Channel channel = (Channel) key;
                        String userId = getUserIdByChannel(channel);
                        log.info("正在解析隧道信息:userId[{}],channel[{}]", userId, JSONUtil.toJsonStr(channel));
                        if (StringUtils.checkUserId(userId, MsgSource.WEB.name())) {
                            sendMessage(channel, message);
                        }
                    });
                }
        );
        log.info("正在往web短推送消息 ---end");
    }

    /**
     * 推送给除了 userId之外的web端
     *
     * @param userIds 排除之外的web端
     * @param message 消息内容
     */
    public static void sendMessageToWebWithoutSpecialUsers(String[] userIds, String message) {
        List<ConcurrentHashMap<Object, Object>> channelUserInfoInstance = UserInstanceCache.getCacheInstance("channelUserInfo");
        channelUserInfoInstance.forEach(
                l -> {
                    l.forEach((key, value) -> {
                        Channel channel = (Channel) key;
                        String cacheUserId = getUserIdByChannel(channel);
                        Arrays.stream(userIds).filter(e -> !StringUtils.checkEqualsUserId(cacheUserId, e)).collect(Collectors.toList()).forEach(
                                e -> {
                                    if (e.contains(MsgSource.WEB.name())) {
                                        sendMessage(channel, message);
                                    }
                                }
                        );
                    });
                }
        );
    }

    /**
     * 推送给 userId之内的的web端
     *
     * @param userIds 包含之内的web端
     * @param message 消息内容
     */
    public static void sendMessageToWebWithinSpecialUsers(String[] userIds, String message) {
        List<ConcurrentHashMap<Object, Object>> channelUserInfoInstance = UserInstanceCache.getCacheInstance("channelUserInfo");
        channelUserInfoInstance.forEach(
                l -> {
                    l.forEach((key, value) -> {
                        Channel channel = (Channel) key;
                        String cacheUserId = getUserIdByChannel(channel);
                        Arrays.stream(userIds).filter(e -> StringUtils.checkEqualsUserId(cacheUserId, e)).collect(Collectors.toList()).forEach(
                                e -> {
//                                    if (e.contains(MsgSource.WEB.name())) {
                                        sendMessage(channel, message);
//                                    }
                                }
                        );
                    });
                }
        );
    }

    /**
     * 推送消息到所有手机端
     *
     * @param message 消息内容
     */
    public static void sendMessageToPhone(String message) {
        List<ConcurrentHashMap<Object, Object>> channelUserInfoInstance = UserInstanceCache.getCacheInstance("channelUserInfo");
        channelUserInfoInstance.forEach(
                l -> {
                    l.forEach((key, value) -> {
                        Channel channel = (Channel) key;
                        String userId = getUserIdByChannel(channel);
                        if (StringUtils.checkUserId(userId, MsgSource.PHONE.name())) {
                            sendMessage(channel, message);
                        }
                    });
                }
        );
    }

    /**
     * 推送给除了 userId之外的安卓端
     *
     * @param userIds 排除之外的安卓端
     * @param message 消息内容
     */
    public static void sendMessageToPhoneWithoutSpecialUsers(String[] userIds, String message) {
        List<ConcurrentHashMap<Object, Object>> channelUserInfoInstance = UserInstanceCache.getCacheInstance("channelUserInfo");
        channelUserInfoInstance.forEach(
                l -> {
                    l.forEach((key, value) -> {
                        Channel channel = (Channel) key;
                        String cacheUserId = getUserIdByChannel(channel);
                        Arrays.stream(userIds).filter(e -> !StringUtils.checkEqualsUserId(cacheUserId, e)).collect(Collectors.toList()).forEach(
                                e -> {
                                    if (e.contains(MsgSource.PHONE.name())) {
                                        sendMessage(channel, message);
                                    }
                                }
                        );
                    });
                }
        );
    }

    /**
     * 推送给特定userId的客户端
     *
     * @param userIds 特定userId
     * @param message 消息内容
     */
    public static void sendMessageToSpecialUsers(String[] userIds, String message) {
        List<ConcurrentHashMap<Object, Object>> channelUserInfoInstance = UserInstanceCache.getCacheInstance("channelUserInfo");
        channelUserInfoInstance.forEach(
                l -> {
                    l.forEach((key, value) -> {
                        Channel channel = (Channel) key;
                        String cacheUserId = getUserIdByChannel(channel);
                        Arrays.stream(userIds).filter(e -> StringUtils.checkEqualsUserId(cacheUserId, e)).collect(Collectors.toList()).forEach(
                                e -> {
                                    sendMessage(channel, message);
                                }
                        );
                    });
                }
        );
    }

    /**
     * 推送给除了特定userId的客户端
     *
     * @param userIds 排除特定userId
     * @param message 消息内容
     */
    public static void sendMessageToWithoutSpecialUsers(String[] userIds, String message) {
        List<ConcurrentHashMap<Object, Object>> channelUserInfoInstance = UserInstanceCache.getCacheInstance("channelUserInfo");
        channelUserInfoInstance.forEach(
                l -> {
                    l.forEach((key, value) -> {
                        Channel channel = (Channel) key;
                        String cacheUserId = getUserIdByChannel(channel);
                        Arrays.stream(userIds).filter(e -> !StringUtils.checkEqualsUserId(cacheUserId, e)).collect(Collectors.toList()).forEach(
                                e -> {
                                    sendMessage(channel, message);
                                }
                        );
                    });
                }
        );
    }


    private static void sendMessage(Channel channel, String message) {
        try {
            //发送队列低于高水位才执行write
            if (channel.isActive() && channel.isWritable()) {
                channel.writeAndFlush(new TextWebSocketFrame(message));
            }
        } catch (Exception e) {
            //发送失败不做处理,防止线程阻塞
        }
    }
}
